package com.catmiao.spark.stream

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

/**
 * @title: SparkStreaming01_WordCount
 * @projectName spark_study
 * @description: TODO
 * @author ChengMiao
 * @date 2024/3/25 00:31
 */
object SparkStreaming02_Queue {

  def main(args: Array[String]): Unit = {


    // 创建环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    //  param1 : 环境配置，SparkConf
    //  param2 ： 采集周期【批量处理周期】
    val ssc = new StreamingContext(sparkConf, Seconds(4))


    // 创建rdd队列
    val rddQueue = new mutable.Queue[RDD[Int]]()
    // 创建 QueueInputDStream
    val inputStream = ssc.queueStream(rddQueue,oneAtATime = false)

    // 处理队列中的rdd数据
    val mappedStream: DStream[(Int, Int)] = inputStream.map((_, 1))
    val reducedStream: DStream[(Int, Int)] = mappedStream.reduceByKey(_ + _)

    // 打印结果
    reducedStream.print()

    // 关闭环境
    // 由于sparkStreaming采集器是长期执行的任务，所以不能直接关闭
    // 如果main方法执行完毕，应用程序也会自动结束，所以不能让main方法执行完毕
    //  ssc.stop()

    // 1. 启动采集器
    ssc.start()

    for (i <- 1 to 5){
      rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)
      Thread.sleep(2000)
    }

    // 2. 等待采集器的关闭
    ssc.awaitTermination()
  }

}
